package com.ryder.subject.infra.basic.es;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.sort.ScoreSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.*;

/**
 * 描述: 自定义的工具类
 *
 * @author Ryder
 * @version 1.0
 * @since 1.0 // 版本号，指明从哪个版本开始
 * 更新日期: 2025/03/10
 */
@Component
@Slf4j
public class EsRestClient {
    // 使用静态 Map 存储不同 Elasticsearch 集群的客户端实例
    // Key 是集群名称，Value 是对应的 RestHighLevelClient
    private static Map<String, RestHighLevelClient> clientMap = new HashMap<>();

    // 自动注入配置类，用于获取 Elasticsearch 集群配置
    @Autowired
    private EsConfigProperties esConfigProperties;

    private static final RequestOptions COMMON_OPTIONS;

    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
        COMMON_OPTIONS = builder.build();
    }

    /**
     * 初始化方法，在 Spring 容器初始化 Bean 后自动调用
     * 用于初始化所有 Elasticsearch 集群的客户端
     */
    @PostConstruct
    public void initialize() {
        // 从配置类中获取 Elasticsearch 集群配置列表
        List<EsClusterConfig> esConfigs = esConfigProperties.getEsConfigs();

        for (EsClusterConfig esConfig : esConfigs) {
            log.info("initialize.config.name:{}, node:{}", esConfig.getName(), esConfig.getNodes());
            // 初始化 RestHighLevelClient
            RestHighLevelClient restHighLevelClient = initRestClient(esConfig);
            // 如果初始化成功，将客户端实例存入 clientMap
            if (restHighLevelClient != null) {
                clientMap.put(esConfig.getName(), restHighLevelClient);
            } else {
                log.error("initialize.config.name:{}, node:{}.initError", esConfig.getName(), esConfig.getNodes());
            }
        }
    }

    /**
     * 初始化 RestHighLevelClient 的方法
     * 根据集群配置创建并返回 RestHighLevelClient 实例
     *
     * @param esClusterConfig 集群配置对象，包含集群名称和节点信息
     * @return RestHighLevelClient 实例，如果初始化失败则返回 null
     */
    private RestHighLevelClient initRestClient(EsClusterConfig esClusterConfig) {
        // 将节点字符串按逗号分隔，获取所有节点
        String[] ipPortArr = esClusterConfig.getNodes().split(",");
        // 用于存储解析后的 HttpHost 对象
        List<HttpHost> httpPostList = new ArrayList<>(ipPortArr.length);
        // 遍历每个节点
        for (String ipPort : ipPortArr) {
            // 将节点按冒号分隔，获取 IP 和端口
            String[] ipPortInfo = ipPort.split(":");
            // 检查节点格式是否正确（IP:端口）
            if (ipPortInfo.length == 2) {
                try {
                    // 解析端口号
                    int port = Integer.parseInt(ipPortInfo[1]);
                    // 创建 HttpHost 对象
                    HttpHost httpHost = new HttpHost(ipPortInfo[0], port);
                    // 将 HttpHost 添加到列表
                    httpPostList.add(httpHost);
                } catch (NumberFormatException e) {
                    log.error("Invalid port format: {}", ipPortInfo[1]);
                }
            }
        }

        if (httpPostList.isEmpty()) {
            log.error("No valid nodes found for cluster: {}", esClusterConfig.getName());
            return null;
        }

        // 将 List<HttpHost> 转换为 HttpHost[]
        HttpHost[] httpHosts = new HttpHost[httpPostList.size()];
        httpPostList.toArray(httpHosts);
        // 使用 HttpHost 数组创建 RestClientBuilder
        RestClientBuilder builder = RestClient.builder(httpHosts);

        // 添加认证信息
        if (StringUtils.isNotBlank(esClusterConfig.getUsername()) && StringUtils.isNotBlank(esClusterConfig.getPassword())) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(
                    AuthScope.ANY,
                    new UsernamePasswordCredentials(esClusterConfig.getUsername(), esClusterConfig.getPassword())
            );

            builder.setHttpClientConfigCallback(httpClientBuilder -> {
                return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            });
        }

        // 创建并返回 RestHighLevelClient
        return new RestHighLevelClient(builder);
    }

    /**
     * 获取指定集群的 RestHighLevelClient 实例
     *
     * @param clusterName 集群名称
     * @return RestHighLevelClient 实例，如果未找到则返回 null
     */
    private static RestHighLevelClient getClient(String clusterName) {
        // 确保 configKey 不为空，且配置正确
        if (clusterName == null) {
            return null; // 这里返回了 null，导致后续问题
        }
        return clientMap.get(clusterName);
    }

    /**
     * 新增文档到 Elasticsearch 索引中
     *
     * @param esIndexInfo  索引信息对象，包含集群名称和索引名称
     * @param esSourceData 文档数据对象，包含文档 ID 和文档内容
     * @return 是否成功插入文档（true 表示成功，false 表示失败）
     */
    public static boolean insertDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            // 创建 IndexRequest 对象，指定索引名称
            IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());
            // 设置文档内容（JSON 格式）
            indexRequest.source(esSourceData.getData());
            // 设置文档 ID（如果未指定，Elasticsearch 会自动生成）
            indexRequest.id(esSourceData.getDocId());
            // 获取指定集群的客户端，并执行插入文档操作
            getClient(esIndexInfo.getClusterName()).index(indexRequest, COMMON_OPTIONS);
            // 插入成功，返回 true
            return true;
        } catch (IOException e) {
            log.error("insertDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 从 Elasticsearch 索引中删除所有文档
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @return 是否成功删除文档（true 表示成功，false 表示失败）
     */
    public static boolean delete(EsIndexInfo esIndexInfo) {
        try {
            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());

            deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());
            BulkByScrollResponse response =
                    getClient(esIndexInfo.getClusterName()).deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);
            long deleted = response.getDeleted();
            log.info("delete.size:{}", deleted);
            return true;
        } catch (Exception e) {
            log.error("delete.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 从 Elasticsearch 索引中删除对应id的文档
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @param docId       文档对应的Id
     * @return 是否成功删除指定文档（true 表示成功，false 表示失败）
     */
    public static boolean deleteDoc(EsIndexInfo esIndexInfo, String docId) {
        try {
            DeleteRequest deleteRequest = new DeleteRequest(esIndexInfo.getIndexName());
            deleteRequest.id(docId);
            DeleteResponse response = getClient(esIndexInfo.getIndexName()).delete(deleteRequest, COMMON_OPTIONS);
            log.info("deleteDoc.size:{}", JSON.toJSONString(response));
            return true;
        } catch (Exception e) {
            log.error("deleteDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }


    /**
     * 修改 Elasticsearch 中的文档
     *
     * @param esIndexInfo  索引信息对象，包含集群名称和索引名称
     * @param esSourceData 文档数据对象，包含文档 ID 和文档内容
     * @return 是否成功修改文档成功（true 表示成功，false 表示失败）
     */
    public static boolean updateDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            UpdateRequest updateRequest = new UpdateRequest();
            updateRequest.index(esIndexInfo.getIndexName());
            updateRequest.id(esSourceData.getDocId());
            updateRequest.doc(esSourceData.getData());
            getClient(esIndexInfo.getClusterName()).update(updateRequest, COMMON_OPTIONS);
            return true;
        } catch (Exception e) {
            log.error("updateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 批量修改 Elasticsearch 中的文档
     *
     * @param esIndexInfo  索引信息对象，包含集群名称和索引名称
     * @param esSourceDataList 文档数据对象，包含文档 ID 和文档内容
     * @return 是否成功修改文档成功（true 表示成功，false 表示失败）
     */
    public static boolean batchUpdateDoc(EsIndexInfo esIndexInfo, List<EsSourceData> esSourceDataList) {
        try {
            boolean flag = false;
            BulkRequest bulkRequest = new BulkRequest();
            for (EsSourceData esSourceData : esSourceDataList) {
                if (StringUtils.isBlank(esSourceData.getDocId())) {
                    continue;
                }
                UpdateRequest updateRequest = new UpdateRequest();
                updateRequest.index(esIndexInfo.getIndexName());
                updateRequest.id(esSourceData.getDocId());
                updateRequest.doc(esSourceData.getData());
                bulkRequest.add(updateRequest);
                flag = true;
            }
            if (flag) {
                BulkResponse bulk = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                return !bulk.hasFailures();
            }
            return true;
        } catch (Exception e) {
            log.error("batchUpdateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 从 Elasticsearch 索引中查询对应id的文档是否存在
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @param docId       文档对应的Id
     * @return 指定文档是否存在（true 表示成功，false 表示失败）
     */
    public static boolean idExistDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            return getClient(esIndexInfo.getIndexName()).exists(getRequest, COMMON_OPTIONS);
        } catch (Exception e) {
            log.error("idExistDocById.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 从 Elasticsearch 索引中查询对应id的文档
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @param docId       文档对应的Id
     * @return 指定id的文档的对象
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            GetResponse response = getClient(esIndexInfo.getIndexName()).get(getRequest, COMMON_OPTIONS);
            return response.getSource();
        } catch (Exception e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 从 Elasticsearch 索引中查询指定字段和id的文档
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @param docId       文档对应的Id
     * @param fields      文档对应的字段
     * @return 指定id和字段相匹配的文档
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo,
                                                 String docId,
                                                 String[] fields) {
        try {
            GetRequest getRequest = new GetRequest(esIndexInfo.getIndexName());
            getRequest.id(docId);
            //使用用FetchSourceContext，设置要返回的字段，需要排除过滤什么
            FetchSourceContext fetchSourceContext = new FetchSourceContext(true, fields, null);
            getRequest.fetchSourceContext(fetchSourceContext);
            GetResponse response = getClient(esIndexInfo.getIndexName()).get(getRequest, COMMON_OPTIONS);
            return response.getSource();
        } catch (Exception e) {
            log.error("getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 从 Elasticsearch 索引 搜索api
     *
     * @param esIndexInfo 索引信息对象，包含集群名称和索引名称
     * @return 结果
     */
    public static SearchResponse searchWithTermQuery(EsIndexInfo esIndexInfo,
                                                     EsSearchRequest esSearchRequest) {
        try {
            BoolQueryBuilder bq = esSearchRequest.getBq();
            String[] fields = esSearchRequest.getFields();
            int from = esSearchRequest.getFrom();
            int size = esSearchRequest.getSize();
            Long minutes = esSearchRequest.getMinutes();
            Boolean needScroll = esSearchRequest.getNeedScroll();
            String sortName = esSearchRequest.getSortName();
            SortOrder sortOrder = esSearchRequest.getSortOrder();

            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(bq);
            searchSourceBuilder.fetchSource(fields, null)
                    .from(from)
                    .size(size);
            //有高亮就存放
            if (Objects.nonNull(esSearchRequest.getHighlightBuilder())) {
                searchSourceBuilder.highlighter(esSearchRequest.getHighlightBuilder());
            }

            if (StringUtils.isNotBlank(sortName)) {
                searchSourceBuilder.sort(sortName);
            }

            searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));

            SearchRequest searchRequest = new SearchRequest();
            searchRequest.searchType(SearchType.DEFAULT);
            searchRequest.indices(esIndexInfo.getIndexName());
            searchRequest.source(searchSourceBuilder);
            if (needScroll) {
                searchRequest.scroll(new Scroll(TimeValue.timeValueMinutes(minutes)));
            }

            return getClient(esIndexInfo.getClusterName()).search(searchRequest, COMMON_OPTIONS);
        } catch (Exception e) {
            log.error("searchWithTermQuery.exception:{}", e.getMessage(), e);
        }
        return null;
    }

}
